package org.softee.messaging.redundancy;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;

/**
 * A filter of messages with unique keys whose purpose it is not to accept consecutive messages with a certain key
 * and identical content ("Redundant Messages").<p/>
 * Example message sequence: (#. key: value)<br/>
 * 1. A: X<br/>
 * 2. A: X<br/>
 * 3. B: Y<br/>
 * 4. A: Z<br/>
 * 5. B: Y<br/>
 * 6. A: X<br/>
 * This filter would remove message numbers 2 and 5.<p/>
 *
 * A history implemented as a LRU cache recalls the most recent value for each key, subject to imposed size and age constraints,
 * which means that no guarantee can be given that all redundant messages will be removed, but guarantees can be given that
 * any non-redundant message will be accepted by this filter.
 *
 * @author morten.hattesen@gmail.com
 *
 * @param <K> The key type
 * @param <V> The value type
 */
public class RedundancyFilter<K, V> {
    private final Cache<K, V> cache;

    public RedundancyFilter(long expiryDuration, TimeUnit expiryUnit, long maximumSize) {
        //FIXME validate arguments
        // TODO consider externalizing cache or allowing non-expiry and no maximum size
        cache = CacheBuilder.newBuilder()
            .expireAfterAccess(expiryDuration, expiryUnit) //TODO consider expireAfterWrite since every access is a write.
            .maximumSize(maximumSize)
            .build();
    }


    public <M extends KeyValueProvider<K, V>> boolean accept(M message) {
        return accept(message.getKey(), message.getValue());
    }

    /**
     * Guarantees:
     * - two consecutive calls with equal key and different value will ALWAYS return true.
     * - two consequtive calls with equal key and equal value will MOSTLY return false (subject to cache expiry and eviction)
     * @param key
     * @param value
     * @return
     */
    boolean accept(K key, V value) {
        V previous = cache.asMap().put(key, value);
        return !eq(value, previous);
    }


    /**
     * Special treatment is given to byte[] value types. Any other array type will result in an exception
     * @param value
     * @param other
     * @return
     * @throws UnsupportedOperationException if value is an array of type other than byte
     */
    private boolean eq(V value, V other) {
        if (value == null || other == null) {
            return value == other;
        }
        if (value == other) {
            return true;
        }
        if (value.getClass().isArray()) {
            if (value instanceof byte[]) {
                return Arrays.equals((byte[])value, (byte[])other);
            } else {
                throw new UnsupportedOperationException("Array type: " + value.getClass().getName());
            }
        } else {
            return value.equals(other);
        }
    }

    public void clear() {
        cache.invalidateAll();
    }
}
